package org.hgf.testrabbitmq.consumer.config;

import org.hgf.testrabbitmq.consumer.handler.ConsumerUnityHandleMessage;
import org.hgf.testrabbitmq.consumer.handler.DeadMessageHandle;
import org.hgf.testrabbitmq.consumer.service.IQueueService;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.Resource;
import java.util.List;

/**
 * 消费者队列配置
 *
 * @author 何国发
 * @version 1.0.0
 * @date 2022/9/7 18:59
 */
@Configuration
public class ConsumerRabbitConfig {

    /**
     * 缓存连接工厂
     */
    @Resource
    private CachingConnectionFactory connectionFactory;

    /**
     * 消费者统一处理消息处理类
     */
    @Resource
    private ConsumerUnityHandleMessage consumerUnityHandleMessage;

    @Resource
    private DeadMessageHandle deadMessageHandle;

    /**
     * 队列服务
     */
    @Resource
    private IQueueService queueService;

    /**
     * 配置消息监听容器
     *
     * @return org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer
     * @author 何国发
     * @date 2022/9/8 10:18
     */
    @Bean("messageListenerContainer")
    public SimpleMessageListenerContainer messageListenerContainer()
    {
        // 初始化消息监听容器
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);

// ------------------------------------- 以下配置是监听全部队列消息的配置 ---------------------------------------------------------------------
        // 获取全部消息队列消息
        List<String> queueNameList = queueService.queueNameList();
        queueNameList.removeIf(DEAD_QUEUE_NAME::equals);
        // 设置队列名称
        container.setQueueNames(queueNameList.toArray(new String[0]));
// --------------------------------------------------------------------------------------------------------------------------------------
        // 开启监听器通道
        container.setExposeListenerChannel(true);
        // 设置监听处理类
        container.setMessageListener(consumerUnityHandleMessage);
        // 设置每个消费者获取的最大的消息数量
        container.setPrefetchCount(1);
        // 设置消费者个数
        container.setConcurrentConsumers(5);
        // 设置确认模式为手动确认
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);

        // 启动容器
        container.start();

        return container;
    }

    /**
     * 死信队列监听容器
     *
     * @return org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer
     * @author 何国发
     * @date 2022/9/9 15:35
     */
    @Bean("deadMessageListenerContainer")
    public SimpleMessageListenerContainer deadMessageListenerContainer()
    {
        // 初始化消息监听容器
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);

        // 设置队列名称
        container.setQueueNames(DEAD_QUEUE_NAME);
        // 开启监听器通道
        container.setExposeListenerChannel(true);
        // 设置监听处理类
        container.setMessageListener(deadMessageHandle);
        // 设置每个消费者获取的最大的消息数量
        container.setPrefetchCount(1);
        // 设置消费者个数
        container.setConcurrentConsumers(5);
        // 设置确认模式为手动确认
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);

        // 启动容器
        container.start();

        return container;
    }

// -------------------------------------------- 死信队列 ------------------------------------------------------------
    /**
     * 死信队列名称
     */
    public static final String DEAD_QUEUE_NAME = "dead_queue";

    /**
     * 死信队列交换机名称
     */
    public static final String DEAD_EXCHANGE_NAME = "dead_exchange";

    /**
     * 路由,fanout类型不需要路由
     */
    public static final String DEAD_ROUTEING = "dead_routeing";

    /**
     * 声明死信队列
     *
     * @return org.springframework.amqp.core.Queue
     * @author 何国发
     * @date 2022/9/9 10:30
     */
    @Bean
    public Queue deadQueue()
    {
        return new Queue(DEAD_QUEUE_NAME);
    }

    /**
     * 声明死信交换机
     *
     * @return org.springframework.amqp.core.Queue
     * @author 何国发
     * @date 2022/9/9 10:30
     */
    @Bean
    public FanoutExchange deadExchange()
    {
        return new FanoutExchange(DEAD_EXCHANGE_NAME);
    }

    /**
     * 建立队列和交换机的绑定关系
     *
     * @return org.springframework.amqp.core.Queue
     * @author 何国发
     * @date 2022/9/9 10:30
     */
    @Bean
    public Binding bindingQueueExchange() {
        return BindingBuilder.bind(deadQueue()).to(deadExchange());
    }
}
